package phoneData

import org.apache.hadoop.conf.Configured
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.io.LongWritable
import org.apache.hadoop.mapreduce.Job
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat
import org.apache.hadoop.util.Tool
import phoneData.MapReduce.{InputMapper, OutputReducer}

class MRDriver extends Configured with Tool {
  override def run(strings: Array[String]): Int = {
    if (strings.length != 2) System.err.println("Error, Usage: yarn jar xxx.jar <input> <output>")

    val conf = getConf
    val fs = FileSystem.get(conf)
    fs.delete(new Path(strings(1)), true)

    val job = Job.getInstance(conf)
    job.setJarByClass(classOf[MRDriver])
    job.setMapperClass(classOf[InputMapper])
    job.setCombinerClass(classOf[OutputReducer])
    job.setReducerClass(classOf[OutputReducer])
    // 设置最终输出结果类型
    job.setOutputKeyClass(classOf[LongWritable])
    job.setOutputValueClass(classOf[FlowBean])
    // 设置输入、输出路径
    FileInputFormat.addInputPath(job, new Path(strings(0)))
    FileOutputFormat.setOutputPath(job, new Path(strings(1)))

    if (job.waitForCompletion(true)) 0 else 1
  }
}
